package com.study.chapter08;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * @Description: 连接流
 * @Author: LiuQun
 * @Date: 2022/8/11 22:04
 */
public class ConnectStreamTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Integer> stream1 = env.fromElements(1, 2, 3, 4);
        DataStreamSource<Long> stream2 = env.fromElements(11L, 22L, 33L, 55L);


        stream1.connect(stream2)
                .map(new CoMapFunction<Integer, Long, String>() {
                    @Override
                    public String map1(Integer value) throws Exception {
                        return "Integer："+value.toString();
                    }

                    @Override
                    public String map2(Long value) throws Exception {
                        return "Long："+value.toString();
                    }
                })
                .print();

        env.execute();
    }
}
